package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
)

//专门忘kafka写日志的模块
var (
	client sarama.SyncProducer //全局的生产者
)

func Init(addrs []string) (err error) {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          //发送完数据需要leader和follow确认
	config.Producer.Partitioner = sarama.NewRandomPartitioner //新选出一个partition
	config.Producer.Return.Successes = true                   //成功交付的消息在success chan 返回

	client, err = sarama.NewSyncProducer(addrs, config)
	if err != nil {
		fmt.Println("error:", err)
	}
	return
}

func SendToKafka(topic string, data string) {
	msg := &sarama.ProducerMessage{}
	msg.Topic = topic
	msg.Value = sarama.StringEncoder(data)
	message, offset, err := client.SendMessage(msg)
	if err != nil {
		fmt.Println("send message error", err)
		return
	}
	fmt.Printf("send success:%v ,offset:%v \n", message, offset)
}
